package cn.edu.ruc.dbiir.htmlparser;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import com.google.common.base.Charsets;

public class HtmlInputFormat extends FileInputFormat<LongWritable, Text>{
	
	
	@Override
	protected boolean isSplitable(JobContext context, Path filename) {
		// TODO Auto-generated method stub
		final CompressionCodec codec =
		          new CompressionCodecFactory(context.getConfiguration()).getCodec(filename);
		        if (null == codec) {
		          return true;
		        }
		        return codec instanceof SplittableCompressionCodec;
	}

	/**
	 * 如何获取分片记录
	 * author：mark   
	 * createTime：Nov 25, 2019 6:13:41 PM
	 * @param split
	 * @param context
	 * @return
	 * @throws IOException
	 * @throws InterruptedException
	 */
	@Override
	public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		String delimiter = context.getConfiguration().get(
		        "textinputformat.record.delimiter");
		    byte[] recordDelimiterBytes = null;
		    if (null != delimiter)
		      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
		    return new HtmlLineRecordReader(recordDelimiterBytes);
	}

}
